package com.hu.wc

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 * @Author: hujianjun
 * @Create Date: 2020/12/2 9:39
 * @Describe: 按省份统计广告点击量
 */
//定义广告日志和输出结果样例类
case class AdClickLog(userId: Long, adId: Long, province: String, city: String, ts: Long)

case class AdClickResult(province: String, windEnd: String, adClickCnt: Long)

object AdClickStatistic {
  def main(args: Array[String]): Unit = {
    val streamEnv = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1，时间语义为事件时间
    streamEnv.setParallelism(1)
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    //读取文件数据源
    val adLogURI = getClass.getResource("/data/adClickLog.csv")
    val inputStream = streamEnv.readTextFile(adLogURI.getPath)

    //转换数据流为样例类流
    val dataStream = inputStream.map(data => {
      val arr = data.split(",")
      AdClickLog(arr(0).toLong, arr(1).toLong, arr(2), arr(3), arr(4).toLong)
    }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[AdClickLog](Time.seconds(3)) {
      override def extractTimestamp(element: AdClickLog): Long = element.ts
    })

    //根据省份分区统计广告点击量
    val resultStream = dataStream.keyBy(_.province)
      .timeWindow(Time.hours(1), Time.seconds(5)) //开滑动窗口，每5s统计过去1小时的省份的点击量
      .aggregate(new PreAggregateADClickCnt(), new ADClickCntResultWindow()) //使用aggregate函数进行统计,第一个自定义函数实现预聚合，第二个输出带有窗口时间结果

    resultStream.print()

    streamEnv.execute("ad click statistic job")
  }
}

class PreAggregateADClickCnt extends AggregateFunction[AdClickLog, Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(in: AdClickLog, acc: Long): Long = acc + 1

  override def getResult(acc: Long): Long = acc

  override def merge(acc: Long, acc1: Long): Long = acc + acc1
}

class ADClickCntResultWindow extends WindowFunction[Long, AdClickResult, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[AdClickResult]): Unit = {
    out.collect(AdClickResult(key, window.getEnd.toString, input.head))
  }
}